package com.gomro.galaxy.foundation.bi.acquisition.job;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.gomro.galaxy.core.batch.ElasticSearchClzItemWriter;
import com.gomro.galaxy.core.commons.elastic.ElasticSearchRest;
import com.gomro.galaxy.foundation.bi.domain.Constants;
import com.gomro.galaxy.foundation.bi.domain.enumerater.AcquireBatchStatus;
import com.gomro.galaxy.foundation.bi.domain.enumerater.AcquirePlan;
import com.gomro.galaxy.foundation.bi.domain.event.BiAcquireBatchEvent;
import com.gomro.galaxy.foundation.bi.domain.model.AcquireLastBatch;
import com.gomro.galaxy.foundation.bi.domain.model.AcquireTrade;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.batch.core.*;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.database.JdbcPagingItemReader;
import org.springframework.batch.item.database.Order;
import org.springframework.batch.item.database.builder.JdbcPagingItemReaderBuilder;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.cloud.bus.BusProperties;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.lang.NonNull;

import javax.sql.DataSource;
import java.util.*;
//交易 贸易
/**
 * @author ZhangLei
 */
@Slf4j
@Configuration
public class TradeJob {

    private static final String BiAcquireLastName = Constants.BiWarehouseIndexPrefix + AcquireLastBatch.TableName;
    private static final String BiWarehouseName = Constants.BiWarehouseIndexPrefix + AcquireTrade.TableName;
    private static final String JobName = Constants.JobBeanNamePrefix + AcquireTrade.TableName;
    private static final String StepName = Constants.StepBeanNamePrefix + AcquireTrade.TableName;
    private static final String ReaderName = Constants.ReaderBeanNamePrefix + AcquireTrade.TableName;


    RowMapper<AcquireTrade> rowMapper = (rs, i) -> AcquireTrade.builder()
            .id(rs.getLong("id"))
            .type(rs.getInt("type"))
            .organizationId(rs.getLong("organization_id"))
            .organizationCompanyName(rs.getString("organization_company_name"))
            .transactionOrganizationId(rs.getLong("transaction_organization_id"))
            .transactionOrganizationCompanyName(rs.getString("transaction_organization_company_name"))
            .memberId(rs.getLong("member_id"))
            .placeOrderMemberFullname(rs.getString("place_order_member_fullname"))
            .placeOrderDepartmentId(rs.getLong("place_order_department_id"))
            .placeOrderDepartmentFullname(rs.getString("place_order_department_fullname"))
            .placeOrderOrganizationId(rs.getLong("place_order_organization_id"))
            .placeOrderOrganizationCompanyName(rs.getString("place_order_organization_company_name"))
            .customerOrderId(rs.getLong("customer_order_id"))
            .vendorOrderId(rs.getLong("vendor_order_id"))
            .orderGoodsId(rs.getLong("order_goods_id"))
            .goldenTax(rs.getString("golden_tax"))
            .num(rs.getDouble("num"))
            .refundAmount(rs.getDouble("refund_amount"))
            .supplementId(rs.getLong("supplement_id"))
            .total(rs.getDouble("total"))
            .freight(rs.getDouble("freight"))
            .platform(rs.getDouble("platform"))
            .platformRate(rs.getDouble("platform_rate"))
            .amount(rs.getDouble("amount"))
            .interest(rs.getDouble("interest"))
            .interestRate(rs.getDouble("interest_rate"))
            .other(rs.getDouble("other"))
            .otherRate(rs.getDouble("other_rate"))
            .accountPeriod(rs.getInt("account_period"))
            .actualAccountPeriod(rs.getInt("actual_account_period"))
            .contract(rs.getBoolean("contract"))
            .alreadyRequestedInvoice(rs.getDouble("already_requested_invoice"))
            .alreadyRequestedInvoiceNum(rs.getDouble("already_requested_invoice_num"))
            .alreadyRequestedPayment(rs.getDouble("already_requested_payment"))
            .redInkAmount(rs.getDouble("red_ink_amount"))
            .redInkNum(rs.getDouble("red_ink_num"))
            .supplementOpenService(rs.getBoolean("supplement_open_service"))
            .invoiceNumber(rs.getString("invoice_number"))
            .accountPeriodExpiry(rs.getBoolean("account_period_expiry"))
            .reconciliationStatus(rs.getInt("reconciliation_status"))
            .reconciliation(rs.getDouble("reconciliation"))
            .payment(rs.getDouble("payment"))
            .invoiceStatus(rs.getInt("invoice_status"))
            .invoiceTime(rs.getTimestamp("invoice_time") == null ? null : rs.getTimestamp("invoice_time").toLocalDateTime())
            .invoice(rs.getDouble("invoice"))
            .reconciliationTime(rs.getTimestamp("reconciliation_time") == null ? null : rs.getTimestamp("reconciliation_time").toLocalDateTime())
            .paymentStatus(rs.getInt("payment_status"))
            .paymentTime(rs.getTimestamp("payment_time") == null ? null : rs.getTimestamp("payment_time").toLocalDateTime())
            .serviceInvoiceStatus(rs.getInt("service_invoice_status"))
            .serviceInvoiceTime(rs.getTimestamp("service_invoice_time") == null ? null : rs.getTimestamp("service_invoice_time").toLocalDateTime())
            .serviceInvoice(rs.getDouble("service_invoice"))
            .refundStatus(rs.getInt("refund_status"))
            .refundTime(rs.getTimestamp("refund_time") == null ? null : rs.getTimestamp("refund_time").toLocalDateTime())
            .refund(rs.getDouble("refund"))
            .refundInvoiceStatus(rs.getInt("refund_invoice_status"))
            .refundInvoiceTime(rs.getTimestamp("refund_invoice_time") == null ? null : rs.getTimestamp("refund_invoice_time").toLocalDateTime())
            .refundInvoice(rs.getDouble("refund_invoice"))
            .refundInvoiceNum(rs.getDouble("refund_invoice_num"))
            .refundServiceInvoiceStatus(rs.getInt("refund_service_invoice_status"))
            .refundServiceInvoiceTime(rs.getTimestamp("refund_service_invoice_time") == null ? null : rs.getTimestamp("refund_service_invoice_time").toLocalDateTime())
            .refundServiceInvoice(rs.getDouble("refund_service_invoice"))
            .summary(rs.getString("summary"))
            .memo(rs.getString("memo"))
            .taxRate(rs.getDouble("tax_rate"))
            .transferPurchase(rs.getBoolean("transfer_purchase"))
            .smallScale(rs.getBoolean("small_scale"))
            .smallScaleRate(rs.getDouble("small_scale_rate"))
            .reconciliationConfirmTime(rs.getTimestamp("reconciliation_confirm_time") == null ? null : rs.getTimestamp("reconciliation_confirm_time").toLocalDateTime())
            .reconciliationDayStrategy(rs.getInt("reconciliation_day_strategy"))
            .organizationPaymentDayStrategy(rs.getInt("organization_payment_day_strategy"))
            .del(rs.getBoolean("del"))
            .last(rs.getTimestamp("last") == null ? rs.getTimestamp("time").toLocalDateTime() : rs.getTimestamp("last").toLocalDateTime())
            .time(rs.getTimestamp("time") == null ? null : rs.getTimestamp("time").toLocalDateTime())
            .createdBy(rs.getString("created_by"))
            .modifiedBy(rs.getString("modified_by"))
            .build();


    private ApplicationEventPublisher applicationEventPublisher;
    private BusProperties busProperties;
    private JobBuilderFactory jobBuilderFactory;
    private StepBuilderFactory stepBuilderFactory;
    private DataSource dataSource;
    private ElasticSearchRest rest;
    private RestHighLevelClient client;
    private ObjectMapper mapper;

    public TradeJob(ApplicationEventPublisher applicationEventPublisher, BusProperties busProperties, JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory, RestHighLevelClient client, ObjectMapper mapper, @Qualifier("gomroDataSource") DataSource dataSource) {

        this.applicationEventPublisher = applicationEventPublisher;
        this.busProperties = busProperties;
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
        this.dataSource = dataSource;

        this.client = client;
        this.mapper = mapper;
        this.rest = new ElasticSearchRest(client, mapper);
    }

    @Bean(JobName)
    public Job job(@Qualifier(StepName) Step step) {

        return jobBuilderFactory.get(JobName)
                .incrementer(new RunIdIncrementer())
                .flow(step)
                .end()
                .listener(jobExecutionListener)
                .build();
    }

    @Bean(StepName)
    public Step step() {

        return stepBuilderFactory.get(StepName)
                .<AcquireTrade, AcquireTrade>chunk(100)
                .reader(reader())
                .processor(processor())
                .writer(writer())
                .build();
    }

    @Bean(ReaderName)
    @StepScope
    public JdbcPagingItemReader<AcquireTrade> reader() {

        // 获取最后更新时间
        AcquireLastBatch acquireLastBatch = rest.documentGet(BiAcquireLastName, AcquireTrade.TableName, AcquireLastBatch.class);
        Date last = null;
        if (acquireLastBatch != null && acquireLastBatch.getLast() != null) {
            last = acquireLastBatch.getLast();
        }
        ;

        String where = null;
        Map<String, Object> parameterValues = new HashMap<>();
        if (last != null) {
            where = "where (last is not null and last >= :last) or (time >= :time)";
            parameterValues.put("last", last);
            parameterValues.put("time", last);
        }

        return new JdbcPagingItemReaderBuilder<AcquireTrade>()
                .name(ReaderName)
                .dataSource(dataSource)
                .pageSize(100)
                .selectClause("select *")
                .fromClause(" from  " + AcquireTrade.TableName)
                .whereClause(where)
                .parameterValues(parameterValues)
                .sortKeys(new LinkedHashMap<String, Order>() {


                    private static final long serialVersionUID = 7416038053283822936L;

                    {
                        put("id", Order.ASCENDING);
                    }
                })
                .rowMapper(rowMapper)
                .build();
    }

    public ElasticSearchClzItemWriter<AcquireTrade> writer() {

        return new ElasticSearchClzItemWriter<>(client, mapper, Constants.BiWarehouseIndexPrefix + AcquireTrade.TableName);
    }

    public ItemProcessor<AcquireTrade, AcquireTrade> processor() {

        return item -> item;
    }

    private JobExecutionListener jobExecutionListener = new JobExecutionListener() {

        @Override
        public void beforeJob(@NonNull JobExecution jobExecution) {

            ElasticSearchRest rest = new ElasticSearchRest(client, mapper);

            // 初始化数据索引
            if (!rest.indexExists(BiWarehouseName)) {

                rest.indexCreate(BiWarehouseName);
                rest.indexMappingPut(BiWarehouseName, "acquisition/" + BiWarehouseName + ".json", true);
                rest.indexFlush(BiWarehouseName, null, null);
            }

            // 初始化最后更新时间索引
            if (!rest.indexExists(BiAcquireLastName)) {

                rest.indexCreate(BiAcquireLastName);
                rest.indexMappingPut(BiAcquireLastName, "acquisition/" + BiAcquireLastName + ".json", true);
                rest.indexFlush(BiAcquireLastName, null, null);
            }
        }

        @Override
        public void afterJob(@NonNull JobExecution jobExecution) {

            // 回写最后更新时间
//            if (jobExecution.getStatus() == BatchStatus.COMPLETED) {
//
//                rest.documentCreate(Constants.BiWarehouseIndexPrefix + AcquireLastBatch.TableName, AcquireTrade.TableName, AcquireLastBatch.builder().id(AcquireTrade.TableName).last(jobExecution.getCreateTime()).build());
//
//                applicationEventPublisher.publishEvent(new BiAcquireBatchEvent(this, busProperties.getId(), "**", AcquirePlan.TradePlan, AcquireBatchStatus.JobCompleted));
//            }
        }
    };
}
